package com.hujf.flink.DataStreamAPI;

import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.IterativeStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @Author hujf
 * @Date 2022/3/17 15:40
 * @Description TODO
 */
public class Iterations {
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        DataStream<String > input = executionEnvironment.readTextFile("./file/word.txt");
//
//        DataStream<Integer> in = input.map((MapFunction<String, Integer>) o -> Integer.valueOf(o));
//
//        IterativeStream<Integer> iteration = in.iterate();
//
//        DataStream<Integer> minusOne = iteration.map(new MapFunction<Integer, Integer>() {
//            @Override
//            public Integer map(Integer value) throws Exception {
//                return value - 1 ;
//            }
//        });
//
//        DataStream<Integer> stillGreaterThanZero = minusOne.filter(new FilterFunction<Integer>() {
//            @Override
//            public boolean filter(Integer value) throws Exception {
//                return (value > 0);
//            }
//        });
//
//        iteration.closeWith(stillGreaterThanZero);
        DataStream<Long> someIntegers = env.generateSequence(0, 1000);

        IterativeStream<Long> iteration = someIntegers.iterate();

        DataStream<Long> minusOne = iteration.map(new MapFunction<Long, Long>() {
            @Override
            public Long map(Long value) throws Exception {
                return value - 1 ;
            }
        });

        DataStream<Long> stillGreaterThanZero = minusOne.filter(new FilterFunction<Long>() {
            @Override
            public boolean filter(Long value) throws Exception {
                return (value > 0);
            }
        });

        iteration.closeWith(stillGreaterThanZero);

        DataStream<Long> lessThanZero = minusOne.filter(new FilterFunction<Long>() {
            @Override
            public boolean filter(Long value) throws Exception {
                return (value <= 0);
            }
        });

    }
}
